package internal

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"go.uber.org/zap"
	"io/ioutil"
	"net/http"
	"gitee.com/carollia/data-sdk-go/model"
	"time"
)

const BATCH_RETRY_COUNT = 3 // 重试3次数后 无效则丢失数据

type Consumer struct {
	JobChannel chan *model.Request
	MaxSize    int
	Quit       chan bool
}

func NewConsumer(JobChannel chan *model.Request, maxSize int) *Consumer {
	return &Consumer{
		JobChannel: JobChannel,
		MaxSize:    maxSize,
		Quit:       make(chan bool),
	}
}

func (w *Consumer) start() {
	go func() {
		defer func() {
			if e := recover(); e != nil {
				Logger.Error("Consumer Get Panic Error", zap.Any("error", e))
				return
			}
		}()
		for {
			select {
			case request := <-w.JobChannel:
				size := len(w.JobChannel)
				// 执行该程序
				if len(request.RequestData.Data) == 0 {
					err := errors.New("consumer start 空数据")
					Logger.Error("consumer start 空数据", zap.String("error", err.Error()), zap.Any("request", request))
					continue
				}
				failIndex, err := DoHttpRequest(request.Url, "POST", request.Timeout, request.RequestData)
				if err == nil {
					// atomic.AddInt32(&global.LogCount, int32(len(request.RequestData.Data)))
					Logger.Debug("收到正确响应:",
						zap.Int("当前队列size", size),
						zap.Int("len(request.RequestData.Data)", len(request.RequestData.Data)),
						zap.Int("failIndex", failIndex),
						zap.Any("request", request), zap.Any("err", err))
				}
				// 失败,队列不满,重试次数小于等于3,队列无堆积
				if err != nil && size <= w.MaxSize-1 && request.RetryCount < BATCH_RETRY_COUNT {
					w.doRetry(request)
				}
			case <-w.Quit:
				Logger.Info("worker服务停止")
				return
			}
		}
	}()
}

// 发送httpPost请求
func DoHttpRequest(url, method string, timeout time.Duration, params model.RequestData) (failIndex int, err error) {
	defer func() {
		if e := recover(); e != nil {
			Logger.Error("DoHttpRequest Get Panic Error", zap.Any("error", e))
			return
		}
	}()
	Logger.Debug("接收参数:", zap.String("url", url), zap.Any("timeout", timeout), zap.Any("params", params))
	jsonStr, err := json.Marshal(params)
	if err != nil {
		return
	}
	req, err := http.NewRequest(method, url, bytes.NewBuffer(jsonStr))
	if err != nil {
		return
	}
	Logger.Debug("组装请求", zap.Any("error", err))
	req.Header.Set("X-Custom-Header", "sdk-go")
	req.Header.Set("Content-Type", "application/json")
	client := &http.Client{Timeout: timeout}
	resp, err := client.Do(req)
	Logger.Debug("收到Resp", zap.Any("error", err))
	if err != nil {
		return
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		err = errors.New(fmt.Sprintf("request is not ok: code: %d", resp.StatusCode))
		return
	}
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return
	}
	respData := new(model.RespData)
	if err = json.Unmarshal(body, respData); err != nil {
		Logger.Error("接收响应:body json解析错误:", zap.String("body", string(body)) , zap.String("error", err.Error()))
		return
	}
	failIndex = respData.Data.FailIndex
	if respData.Code != 0 {
		err = errors.New(fmt.Sprintf("错误码:%d, 错误信息:%s", respData.Code, respData.Message))
		return
	}
	Logger.Debug("响应结果", zap.Any("respData", respData))
	return
}

// 重试任务
func (w *Consumer) doRetry(request *model.Request) {
	request.RetryCount++
	// 10秒后加入请求
	//time.Sleep(30 * time.Second)
	go func(req *model.Request) {
		Logger.Debug("请求服务失败,开始重试",
			zap.Int("当前队列size", len(w.JobChannel)), zap.Int("当前重试次数", request.RetryCount),
			zap.Int("len(request.RequestData.Data)", len(request.RequestData.Data)), zap.Any("request", request))
		w.JobChannel <- req
	}(request)
}

// 停止
func (w *Consumer) stop() {
	go func() {
		w.Quit <- true
	}()
}

